﻿using Furion;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using VOL.RabbitMq.IRabbitMq;
using VOL.RabbitMq.RabbitMqDto;

namespace VOL.RabbitMq.RabbitMq
{
    /// <summary>
    /// MQTT通讯客户端
    /// </summary>
    public class RabbitMqClient : IRabbitMqClient
    {
        #region 变量
        /// <summary>
        /// 客户端连接属性
        /// </summary>
        private RabbitMqModel mCommOptions =new RabbitMqModel();


        /// <summary>
        /// 连接工厂类
        /// </summary>
        private IConnectionFactory mClientOptions;
        /// <summary>
        /// 连接信息
        /// </summary>
        private IConnection connection;

        /// <summary>
        /// 消费者
        /// </summary>
        //new AsyncEventingBasicConsumer(channel)
        private EventingBasicConsumer basicConsumer;


        /// <summary>
        /// 控制器选项列表
        /// </summary>
        private readonly List<HandlerDto> mHandlerOptions;

        /// <summary>
        /// 通道信息
        /// </summary>
        private IModel model;

        /// <summary>
        /// 订阅客户端状态主题
        /// </summary>
        private readonly string mSubClientStateTopic;
        /// <summary>
        /// 是否重连
        /// </summary>
        private Boolean mIsReconnect = false;

        #endregion

        #region 属性

        /// <summary>
        /// MQTT通讯客户端标识
        /// </summary>
        public string ClientId => mCommOptions.RabbitOptions.ClientId;

        /// <summary>
        /// 客户端编码
        /// </summary>
        public string ClientCode => mCommOptions.RabbitOptions.ClientId;

        /// <summary>
        /// MQTT通讯客户端连接状态
        /// </summary>
        public Boolean IsConnected => connection.IsOpen;


        /// <summary>
        /// 是否关闭
        /// </summary>
        public Boolean IsClosed { get; private set; } = false;

        #endregion

        #region 构造函数

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="provider">服务提供者</param>
        public RabbitMqClient(IServiceProvider provider)
        {
            var commOptions = App.GetConfig<RabbitMqModel>("RabbitMq", true);

            mCommOptions = commOptions;
            //// 服务提供者
            //mProvider = provider;
            // Rabbit连接建立
            mClientOptions = ClientOptionBuilder();

            //创建初始连接
            connection = mClientOptions.CreateConnection(this.ClientCode);
            //初始创建通道
            model = this.AddChannelAsync().Result;


            //初始化构造时清除所有旧信息
            this.ClearAllMsg("Sorting", "Sorting.InBagPoint");
            this.ClearAllMsg("Sorting", "Sorting.EndBagPoint");
            this.ClearAllMsg("Sorting", "Sorting.SendToLoc");
            this.ClearAllMsg("Sorting", "Sorting.SendFinish");

            this.ClearAllMsg("SortingView", "SortingView.EquInitial");
            this.ClearAllMsg("SortingView", "SortingView.Error");
            this.ClearAllMsg("SortingView", "SortingView.PutCount");
            this.ClearAllMsg("SortingView", "SortingView.CountFinish");
            this.ClearAllMsg("SortingView", "SortingView.StartStopTask");
            this.ClearAllMsg("SortingView", "SortingView.SendStartKeyout");


            //创建消费者
            this.basicConsumer = new EventingBasicConsumer(model);
            mHandlerOptions = commOptions.RabbitMqHandler;
            // 客户端断开事件
            connection.CallbackException += (sender, e) =>
            {
                if (mIsReconnect) return;
                if (!IsClosed)
                {
                    mIsReconnect = true;
                    System.Diagnostics.Debug.WriteLine($"客户端({ClientId})断开重连MQTT服务器，3秒钟后再次重连");
                    Console.WriteLine($"客户端({ClientId})断开重连MQTT服务器，3秒钟后再次重连");
                    while (true)
                    {
                        Task.Delay(TimeSpan.FromSeconds(3)).Wait();
                        if (this.ReConnect())
                        {
                            Console.WriteLine($"客户端({ClientId})重连成功");
                            mIsReconnect = false;
                            break;
                        }
                        Console.WriteLine($"客户端({ClientId})重连失败，3秒钟后再次重连");
                    }
                }
                else
                    Console.WriteLine($"客户端({ClientId})断开MQTT服务器");
            };
            // 客户端连接事件
            if (connection.IsOpen)
            {
                System.Diagnostics.Debug.WriteLine($"客户端({ClientId})成功连接RabbitMq服务器");
                Console.Out.WriteLineAsync($"客户端({ClientId})成功连接RabbitMq服务器");
            }


            // 客户端接收消息事件
            basicConsumer.Received += (a, args) =>
            {

                var message = Encoding.UTF8.GetString(args.Body.Span);//接收到的消息
                // 输出客户端处理消息
                System.Diagnostics.Debug.WriteLine($"客户端({ClientId})接收到订阅数据：{args.Exchange}-{args.RoutingKey}-{message}，时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");
                //Console.WriteLine($"客户端({ClientId})接收到订阅数据：{args.Exchange}-{args.RoutingKey}-{message}，时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");
                // 转换机器人负载数据
                BasePayload payload = null;
                payload = new BasePayload { Handler = "Handler.RabbitMqMessageHandler", IsSys = true };
                if (args.Exchange == "Sorting")
                {
                    payload = new BasePayload { Handler = "Handler.RabbitMqMessageHandler", IsSys = true };

                }
                else if (args.RoutingKey == "SortingView.Error")
                {

                    payload = new BasePayload { Handler = "Views.ViewTaskMan", IsSys = true };

                }

                if (payload == null) return;
                // 获取处理器
                HandlerDto handler = mHandlerOptions.FirstOrDefault(y => y.Handler == payload.Handler);
                if (handler == null) return;

                Task.Delay(TimeSpan.FromMilliseconds(100)).Wait();
                try
                {
                    var services = (IRabbitMqMessageHandler)provider.GetRequiredService(handler.HandlerType);
                    var result = services
                          .TopicPayloadAsyncHandler(model, args.Exchange,
                              args.RoutingKey, message).Result;
                    if (result)
                    {
                        //返回消息确认
                        model.BasicAck(args.DeliveryTag, true);
                        Console.ForegroundColor = ConsoleColor.White;
                        Console.WriteLine($"客户端({ClientId})处理成功:{args.Exchange}-{args.RoutingKey + message}，时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");
                        System.Diagnostics.Debug.WriteLine($"客户端({ClientId})处理成功:{args.Exchange}-{args.RoutingKey + message}，时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");
                    }
                    else
                    {
                        //返回消息确认 重新放回队列
                        model.BasicNack(args.DeliveryTag, false, true);
                        //抛弃此条消息
                        //channel.basicNack(envelope.getDeliveryTag(), false, false);
                    }
                }
                catch (Exception ex)
                {
                    Console.ForegroundColor = ConsoleColor.Red;
                    Console.WriteLine(ex.ToString());
                    throw ex;
                }

                //Console.WriteLine($"客户端({ClientId})处理完成:{args.Exchange}-{args.RoutingKey + message}，时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");

            };
            if (commOptions == null)
            {
                // mCommOptions = provider
            }
            //configuration.GetSection("MqttClient")
        }

        #endregion

        #region 接口实现

        /// <summary>
        ///创建连接
        /// </summary>
        /// <returns></returns>
        public bool ReConnect()
        {

            try
            {
                // 如果客户端处于未连接状态，进行连接
                if (!IsConnected)
                {
                    mClientOptions = ClientOptionBuilder();
                    connection = mClientOptions.CreateConnection(this.ClientCode);

                }
                // 返回连接状态
                return connection.IsOpen;
            }
            catch (Exception)
            {
                return false;
            }
            finally
            {
                // 设置关闭
                IsClosed = false;
            }
        }


        /// <summary>
        /// 创建连接
        /// </summary>
        /// <returns></returns>
        public async Task<bool> ConnectAsync()
        {
            int index = 0;
            try
            {
                // 如果客户端处于未连接状态，进行连接
                if (!IsConnected)
                {
                    // 如果连接成功
                    if (this.ReConnect())
                    {
                        return await Task.FromResult(true);
                    }
                    else
                    {
                        //失败时执行三次重连尝试
                        while (index <= 3)
                        {
                            Task.Delay(TimeSpan.FromSeconds(3)).Wait();
                            await this.ConnectAsync();  //再次调用连接
                            index++;
                        }

                    }
                }
                // 返回连接状态
                return await Task.FromResult(IsConnected);
            }
            catch (Exception)
            {
                return await Task.FromResult(false);
            }
            finally
            {

                IsClosed = false;

            }

        }

        /// <summary>
        /// 关闭连接
        /// </summary>
        /// <returns></returns>

        public bool Disconnect()
        {
            try
            {
                if (connection == null)
                {
                    return false;

                }
                if (connection.IsOpen)
                {
                    connection.Close();
                    if (connection.IsOpen)
                    {
                        return false;
                    }
                }
                return true;

            }
            catch (Exception)
            {
                return false;
                throw;
            }
        }

        /// <summary>
        /// 关闭连接
        /// </summary>
        /// <returns></returns>

        public Task<bool> DisconnectAsync()
        {
            try
            {
                if (connection == null)
                {
                    return Task.FromResult(false);

                }
                if (connection.IsOpen)
                {
                    connection.Close();
                    if (connection.IsOpen)
                    {
                        return Task.FromResult(false);
                    }
                }
                return Task.FromResult(true);

            }
            catch (Exception)
            {
                return Task.FromResult(false);
                throw;
            }
        }



        /// <summary>
        /// 创建通道
        /// </summary>
        /// <returns></returns>
        public Task<IModel> AddChannelAsync()
        {
            try
            {
                if (model == null)
                {
                    model = connection.CreateModel();
                    return Task.FromResult(model);
                }
                else
                {
                    return Task.FromResult(model);
                }

            }
            catch (Exception)
            {

                throw;
            }
        }
        /// <summary>
        /// 创建交换机
        /// </summary>
        /// <param name="ExchangeType"></param>
        /// <returns></returns>
        public bool AddeExchange(string ExchangeName, string ExchangeType)
        {
            try
            {
                if (string.IsNullOrEmpty(ExchangeName))
                {
                    return false;
                }
                if (string.IsNullOrEmpty(ExchangeType))
                {
                    return false;
                }
                if (!connection.IsOpen)
                {
                    return false;
                }
                if (model == null)
                {
                    return false;
                }

                model.ExchangeDeclare(exchange: ExchangeName, type: ExchangeType, durable: true, autoDelete: false, null);   //创建Exchange

                return true;

            }
            catch (Exception)
            {
                return false;
                throw;
            }
        }

        public Task<bool> AddeExchangeAsync(string ExchangeName, string ExchangeType)
        {
            try
            {
                if (string.IsNullOrEmpty(ExchangeName))
                {
                    return Task.FromResult(false);
                }
                if (string.IsNullOrEmpty(ExchangeType))
                {
                    return Task.FromResult(false);
                }
                if (!connection.IsOpen)
                {
                    return Task.FromResult(false);
                }
                var channel = connection.CreateModel();
                channel.ExchangeDeclare(exchange: ExchangeName, type: ExchangeType, durable: true, autoDelete: false, null);   //创建Exchange

                return Task.FromResult(true);

            }
            catch (Exception)
            {
                return Task.FromResult(false); ;
                throw;
            }
        }



        /// <summary>
        /// 删除交换机
        /// </summary>
        /// <param name="ExchangeName"></param>
        /// <returns></returns>

        public bool DelExchange(string ExchangeName)
        {
            try
            {
                if (string.IsNullOrEmpty(ExchangeName))
                {
                    return false;
                }
                if (!connection.IsOpen)
                {
                    return false;
                }
                model.ExchangeDelete(exchange: ExchangeName, false);

                return true;

            }
            catch (Exception)
            {
                return false;
                throw;
            }
        }

        /// <summary>
        /// 删除交换机
        /// </summary>
        /// <param name="ExchangeName"></param>
        /// <returns></returns>

        public Task<bool> DelExchangeAsync(string ExchangeName)
        {
            try
            {
                if (string.IsNullOrEmpty(ExchangeName))
                {
                    return Task.FromResult(false);
                }
                if (!connection.IsOpen)
                {
                    return Task.FromResult(false);
                }
                model.ExchangeDelete(exchange: ExchangeName, false);
                return Task.FromResult(true);

            }
            catch (Exception)
            {
                return Task.FromResult(false);
                throw;
            }
        }


        /// <summary>
        /// 创建队列s
        /// </summary>
        /// <param name="queuename"></param>
        /// <returns></returns>
        public bool AddQueue(string queuename)
        {
            try
            {
                if (string.IsNullOrEmpty(queuename))
                {
                    return false;
                }
                if (!connection.IsOpen)
                {
                    return false;
                }
                model.QueueDeclare(queue: queuename, true, false, false);

                return true;

            }
            catch (Exception)
            {
                return false;
                throw;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="queuename"></param>
        /// <param name="durable"></param>
        /// <returns></returns>
        public bool AddQueue(string queuename, bool durable)
        {
            try
            {
                if (string.IsNullOrEmpty(queuename))
                {
                    return false;
                }
                if (!connection.IsOpen)
                {
                    return false;
                }
                model.QueueDeclare(queue: queuename, durable, false, false);

                return true;

            }
            catch (Exception)
            {
                return false;
                throw;
            }
        }

        /// <summary>
        /// 创建队列
        /// </summary>
        /// <param name="queuename"></param>
        /// <param name="durable"></param>
        /// <param name="exclusive"></param>
        /// <param name="autoDelete"></param>
        /// <returns></returns>

        public bool AddQueue(string queuename, bool durable, bool exclusive, bool autoDelete)
        {
            try
            {
                if (string.IsNullOrEmpty(queuename))
                {
                    return false;
                }
                if (!connection.IsOpen)
                {
                    return false;
                }
                model.QueueDeclare(queue: queuename, durable, exclusive, autoDelete);

                return true;

            }
            catch (Exception)
            {
                return false;
                throw;
            }
        }
        /// <summary>
        /// 删除队列
        /// </summary>
        /// <param name="queuename"></param>
        /// <returns></returns>

        public bool DelQueue(string queuename)
        {
            try
            {
                if (string.IsNullOrEmpty(queuename))
                {
                    return false;
                }
                if (!connection.IsOpen)
                {
                    return false;
                }
                model.QueueDelete(queue: queuename, false);

                return true;

            }
            catch (Exception)
            {
                return false;
                throw;
            }
        }


        /// <summary>
        /// 清除所有消息
        /// </summary>

        public void ClearAllMsg(string exchangeName, string routingKey)
        {
            this.AddeExchange(exchangeName, "topic");//创建交换机
            this.AddQueue(routingKey); //创建队列

            model.QueueBind(queue: routingKey, exchange: exchangeName, routingKey); //绑定交换机和队列
            var properties = model.CreateBasicProperties();
            properties.DeliveryMode = 2;//持久化

            model.QueuePurge(routingKey);
        }


        /// <summary>
        /// 发布
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="exchangeName"></param>
        /// <param name="routingKey"></param>
        /// <param name="payload"></param>
        /// <returns></returns>
        public Task<RabbitMqResultDto> PublishAsync(string exchangeName, string routingKey, RabbitSendDto<dynamic> payload)
        {
            System.Threading.Thread.Sleep(10);
            int index = 3;
            RabbitMqResultDto RabbitMqResultDto = new RabbitMqResultDto();
            RabbitMqResultDto.Statue = true;
            if (!connection.IsOpen)
            {
                while (index <= 3)
                {
                    var res = this.ReConnect();
                    if (res)
                    {
                        break;
                    }
                    index++;
                }
                RabbitMqResultDto.Statue = false;
                RabbitMqResultDto.Message = "RabbitMq服务器连接失败！";
                return Task.FromResult(RabbitMqResultDto);
            }

            try
            {



                model.ConfirmSelect(); //开启服务端确认
                //这两个事件如果是using写法要写到创建交换器和队列并且进行绑定之前，否则不会触发
                model.BasicNacks += (sender, e) =>
                {
                    //生产者发送消息到broker（服务器）后失败被生产者的listener监听到，就走无应答方法
                    Console.WriteLine(" --no ack-- " + e.ToString());
                };
                //有应答
                model.BasicAcks += (sender, e) =>
                {
                    Console.WriteLine(" --ack-- " + e.ToString());
                };


                this.AddeExchange(exchangeName, "topic");//创建交换机
                this.AddQueue(routingKey); //创建队列

                model.QueueBind(queue: routingKey, exchange: exchangeName, routingKey); //绑定交换机和队列
                var properties = model.CreateBasicProperties();
                properties.DeliveryMode = 2;//持久化

                var msgContent = JsonConvert.SerializeObject(payload);
                var msgByte = Encoding.UTF8.GetBytes(msgContent);

                model.BasicPublish
                (
                    exchange: exchangeName,
                    routingKey: routingKey,
                    basicProperties: properties,
                    body: msgByte
                );
                if (!model.WaitForConfirms())
                {  //broker confirm后producer调用
                    RabbitMqResultDto.Statue = false;
                    RabbitMqResultDto.Message = "消息从Rms服务器至RabbiotMq服务器发送失败！未成功写入队列！";
                    return Task.FromResult(RabbitMqResultDto);
                }
                System.Threading.Thread.Sleep(10);
                return Task.FromResult(RabbitMqResultDto);
            }
            catch (Exception ex)
            {
                model.TxRollback();
                Console.ForegroundColor=ConsoleColor.Red;
                Console.WriteLine(ex.ToString());
                return Task.FromResult(RabbitMqResultDto);
                throw;
            }
        }


        /// <summary>
        /// 订阅消费
        /// </summary>
        /// <param name="exchangeName"></param>
        /// <param name="routingKey"></param>
        /// <param name="autoAck"></param>
        /// <returns></returns>
        public Task<bool> SubscribeAsync(string exchangeName, string routingKey, bool autoAck = false)
        {
            try
            {
                this.AddeExchange(exchangeName, "topic");//创建交换机
                this.AddQueue(routingKey); //创建队列

                model.QueueBind(queue: routingKey, exchange: exchangeName, routingKey); //绑定交换机和队列
                var properties = model.CreateBasicProperties();
                properties.DeliveryMode = 2;//持久化


                //声明为手动确认，每次只消费1条消息。
                model.BasicQos(0, 1, false);
                //开启监听
                model.BasicConsume(queue: routingKey, autoAck: autoAck, consumer: basicConsumer);

                return Task.FromResult(true);
            }
            catch (Exception)
            {
                return Task.FromResult(false);
                throw;
            }
        }

        /// <summary>
        /// 取消订阅
        /// </summary>
        /// <param name="exchangeName"></param>
        /// <param name="routingKey"></param>
        /// <returns></returns>

        public Task<bool> UnsubscribeAsync(string exchangeName, string routingKey)
        {
            throw new NotImplementedException();
        }

        /// <summary>
        /// 获取指定交换机中的数量
        /// </summary>
        /// <param name="exchangeName"></param>
        /// <param name="routingKey"></param>
        /// <returns></returns>
        public  Task<uint> GetCount(string exchangeName, string routingKey)
        {

            var res = model.MessageCount(routingKey);

            return Task.FromResult(res);
        }
        #endregion



        #region 构造通讯客户端选项

        /// <summary>
        /// 构造通讯连接选项
        /// </summary>
        /// <returns></returns>
        private IConnectionFactory ClientOptionBuilder()
        {
            var _ = (mCommOptions
                ?? throw new ArgumentNullException("mCommOptions", "连接属性不可为空"));
            var _config = mCommOptions;
            ConnectionFactory factory = new ConnectionFactory();
            factory.UserName = _config.RabbitOptions.UserName; // "gue 
            factory.Password = _config.RabbitOptions.Password;
            factory.HostName = _config.RabbitOptions.Server;
            factory.Port = _config.RabbitOptions.Port;
            factory.AutomaticRecoveryEnabled = true;
            return factory;
        }


        #endregion
    }
}
